package io.sqrtqiezi.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object RDDQueueDStream {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("spark socket streaming")
      .setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(1))

    val rddQueue = new mutable.Queue[RDD[Int]]
    val queueStream = ssc.queueStream(rddQueue)
    val mappedStream = queueStream.map(r => (r % 10, 1))
    val reducedStream = mappedStream.reduceByKey(_ + _)

    reducedStream.print
    ssc.start

    for (i <- 1 to 10) {
      rddQueue.synchronized {
        val range = (1 to 100).map(_ * i)
        rddQueue += ssc.sparkContext.makeRDD(range, 2)
      }
      Thread.sleep(2000)
    }

    ssc.stop()
  }
}
